package com.group1.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.app.BaseAppV2;
import com.group1.realtime.bean.TradeOrderPersonNumBean;
import com.group1.realtime.util.AtguiguUtil;
import com.group1.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.JoinedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

import static com.group1.realtime.common.Constant.*;

public class Dws_Trade_OrderPersonNum extends BaseAppV2 {

    public static void main(String[] args) {
        new Dws_Trade_OrderPersonNum().init(
                3333,
                2,
                "Dws_Transaction_OrderPersonNum",
                TOPIC_DWD_TRAFFIC_PAGE,
                TOPIC_ODS_DB,
                TOPIC_DWD_TRAFFIC_UV_DETAIL
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env,
                          HashMap<String, DataStreamSource<String>> topicAndStreamMap) {

        //双流join，获取字段
        DataStream<TradeOrderPersonNumBean> beanStream = parseAndUnionOne(topicAndStreamMap);
        beanStream.print();
        // 按照uid进行去重
        SingleOutputStreamOperator<TradeOrderPersonNumBean> distinctStream = distinctByUid(beanStream);
        //按照sc分组, 开窗聚合
        SingleOutputStreamOperator<TradeOrderPersonNumBean> resultStream = windowAndAgg(distinctStream);
        //resultStream.print();
        //数据写出到clickhouse中
        writeToClickHouse(resultStream);
    }

    private void writeToClickHouse(SingleOutputStreamOperator<TradeOrderPersonNumBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink("", TradeOrderPersonNumBean.class));
    }

    private SingleOutputStreamOperator<TradeOrderPersonNumBean> windowAndAgg(SingleOutputStreamOperator<TradeOrderPersonNumBean> distinctStream) {
        return distinctStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                        .<TradeOrderPersonNumBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean,ts) -> bean.getTs())
                )
                .keyBy(TradeOrderPersonNumBean::getSc)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeOrderPersonNumBean>() {
                    @Override
                    public TradeOrderPersonNumBean reduce(TradeOrderPersonNumBean value1,
                                                          TradeOrderPersonNumBean value2) throws Exception {
                        value1.setOrder_num(value1.getOrder_num() + value2.getOrder_num());
                        value1.setUser_num(value1.getUser_num() + value2.getUser_num());
                        return value1;
                    }
                }, new ProcessWindowFunction<TradeOrderPersonNumBean, TradeOrderPersonNumBean, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<TradeOrderPersonNumBean> elements,
                                        Collector<TradeOrderPersonNumBean> out) throws Exception {

                        TradeOrderPersonNumBean bean = elements.iterator().next();

                        bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                        bean.setTranfrom_rate((double) (bean.getOrder_num() / bean.getUser_num()));
                        bean.setTs(ctx.currentProcessingTime());

                        out.collect(bean);
                    }
                });


    }

    private SingleOutputStreamOperator<TradeOrderPersonNumBean> distinctByUid(DataStream<TradeOrderPersonNumBean> beanStream) {

        return beanStream
                .keyBy(TradeOrderPersonNumBean::getUser_id)
                .process(new KeyedProcessFunction<String, TradeOrderPersonNumBean, TradeOrderPersonNumBean>() {

                    private ValueState<TradeOrderPersonNumBean> singleUidState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        singleUidState = getRuntimeContext().getState(new ValueStateDescriptor<TradeOrderPersonNumBean>("singleUidState", TradeOrderPersonNumBean.class));
                    }

                    @Override
                    public void processElement(TradeOrderPersonNumBean value,
                                               Context ctx,
                                               Collector<TradeOrderPersonNumBean> out) throws Exception {

                        //第一条数据进来的时候, 状态应该是空, 更新状态
                        if (singleUidState.value() == null) {
                            // 1. 更新状态
                            singleUidState.update(value);

                            out.collect(singleUidState.value());
                        }

                    }
                });
    }

    private DataStream<TradeOrderPersonNumBean> parseAndUnionOne(Map<String, DataStreamSource<String>> topicAndStreamMap) {

        SingleOutputStreamOperator<TradeOrderPersonNumBean> ods_dbStream = topicAndStreamMap
                .get(TOPIC_ODS_DB)
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                             return "order_info".equals(JSON.parseObject(value).getString("table"));

                    }
                })
                .map(new MapFunction<String, TradeOrderPersonNumBean>() {
                    @Override
                    public TradeOrderPersonNumBean map(String value) throws Exception {

                        JSONObject obj = JSON.parseObject(value);
                        String user_id = obj.getJSONObject("data").getString("user_id");
                        String order_id = obj.getJSONObject("data").getString("id");
                        Long ts = obj.getLong("ts") * 1000;
                        Long order_num = 1L;

                        return new TradeOrderPersonNumBean(
                                "", "",
                                user_id,
                                "",
                                order_num,
                                0L,
                                order_id,
                                0.0,
                                ts
                        );
                    }
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TradeOrderPersonNumBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                );



        SingleOutputStreamOperator<TradeOrderPersonNumBean> pageStream = topicAndStreamMap
                .get(TOPIC_DWD_TRAFFIC_PAGE)
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {

                        JSONObject obj = JSON.parseObject(value);

                        return "order".equals(obj.getJSONObject("page").getString("page_id"))
                                && obj.getJSONObject("page").containsKey("item_type")
                                && "order_id".equals(obj.getJSONObject("page").getString("item_type"));
                    }
                })
                .map(new MapFunction<String, TradeOrderPersonNumBean>() {
                    @Override
                    public TradeOrderPersonNumBean map(String value) throws Exception {

                        JSONObject obj = JSON.parseObject(value);
                        JSONObject common = obj.getJSONObject("common");


                        Long ts = obj.getLong("ts");
                        String sc = common.getString("sc");
                        String order_id = obj.getJSONObject("page").getString("item");

                        return new TradeOrderPersonNumBean(
                                "", "",
                                "",
                                sc,
                                0L,
                                0L,
                                order_id,
                                0.0,
                                ts
                        );
                    }
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TradeOrderPersonNumBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                );


        SingleOutputStreamOperator<TradeOrderPersonNumBean> UVStream = topicAndStreamMap
                .get(TOPIC_DWD_TRAFFIC_UV_DETAIL)
                /*.filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        JSONObject obj = JSON.parseObject(value);

                        return "order".equals(obj.getJSONObject("page").getString("page_id"))
                                || "order".equals(obj.getJSONObject("page").getString("last_page_id"));

                    }
                })*/
                .map(new MapFunction<String, TradeOrderPersonNumBean>() {
                    @Override
                    public TradeOrderPersonNumBean map(String value) throws Exception {

                        JSONObject obj = JSON.parseObject(value);
                        String sc = obj.getJSONObject("common").getString("sc");

                        Long ts = obj.getLong("ts");

                        //String order_id = obj.getJSONObject("page").getString("item");

                        return new TradeOrderPersonNumBean(
                                "", "",
                                "",
                                sc,
                                0L,
                                1L,
                                "",
                                0.0,
                                ts
                        );
                    }
                });

        return ods_dbStream
                .keyBy(TradeOrderPersonNumBean::getOrder_id)
                .intervalJoin(pageStream.keyBy(TradeOrderPersonNumBean::getOrder_id))
                .between(Time.seconds(-2), Time.seconds(3))
                .process(new ProcessJoinFunction<TradeOrderPersonNumBean, TradeOrderPersonNumBean, TradeOrderPersonNumBean>() {
                    @Override
                    public void processElement(TradeOrderPersonNumBean first, TradeOrderPersonNumBean second, Context context, Collector<TradeOrderPersonNumBean> collector) throws Exception {
                        collector.collect(new TradeOrderPersonNumBean("", "", first.getUser_id(),second.getSc(),first.getOrder_num(),0L,first.getOrder_id(),0.0,first.getTs()));
                    }
                }).union(UVStream);


    }


}
